package com.baishancloud.log

import com.baishancloud.log.common.util.TimeUtil
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.util.{Timer, TimerTask}

/**
 * 转换数据为对象，过滤延迟数据
 *
 * @author ziqiang.wang
 * @date 2021/12/8 09:01
 */
class MiguICTAuditLogParse(parameterTool: ParameterTool) extends ProcessFunction[Array[String], MiguICTAuditLog] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[MiguICTAuditLogParse])
  //允许最迟数据时间
  private var allowOldest: Long = -1

  override def open(parameters: Configuration): Unit = {
    new Timer("定时更新允许最迟时间", true).scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = allowOldest = System.currentTimeMillis() - 30 * 24 * 3600 * 1000L
    }, 0, 3600 * 1000)
  }

  override def processElement(value: Array[String], ctx: ProcessFunction[Array[String], MiguICTAuditLog]#Context, out: Collector[MiguICTAuditLog]): Unit = {
    val logMillisecond: Long = TimeUtil.stringToLong(value(0), "yyyyMMdd'T'HHmmss'Z'", 8)
    if (logMillisecond >= allowOldest) {
      out.collect(
        MiguICTAuditLog(
          TimeUtil.timeAtString(logMillisecond, parameterTool.getLong(aggInterval, 5) * 60 * 1000),
          value(6),
          1L,
          value(14).toLong)
      )
    }
  }
}
